package com.burges.net.dataStream.waterMark

import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.watermark.Watermark

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/1/27 
  * 描述     用户自定义实现PeriodicWatermarks
  */
class DefinedPeriodicWatermarks extends AssignerWithPeriodicWatermarks[(String, Int, Int)]{

	//1s演示设定，表示在1s以内的数据延时有效，超过一秒的数据被认定为迟到时间
	val maxOutOfOrderness = 1000L
	var currentMaxTimestamp: Long = _

	//覆写该方法，生成watermark
	override def getCurrentWatermark: Watermark = {
		//根据最大事件时间减去最大的乱序时延长度，然后获得取值
		new Watermark(currentMaxTimestamp - maxOutOfOrderness)
	}

	def max(currentTimestamp: Long, currentMaxTimestamp: Long): Long = {
		if(currentMaxTimestamp >= currentTimestamp)
			return currentMaxTimestamp
		else
			return currentTimestamp
	}

	override def extractTimestamp(event : (String, Int, Int), l: Long): Long = {
		//复写currentTimestamp方法，获取当前事件时间
		val currentTimestamp = event._2
		//对比当前事件时间和历史最大事件时间，重新赋值
		currentMaxTimestamp = max(currentTimestamp, currentMaxTimestamp)
		currentTimestamp
	}
}
